package com.ccf.business.etl.service.impl;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.date.DateField;
import cn.hutool.core.date.DateTime;
import cn.hutool.core.date.DateUtil;
import cn.hutool.core.util.IdUtil;
import cn.hutool.core.util.ObjectUtil;
import cn.hutool.core.util.StrUtil;
import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.ccf.business.etl.model.*;
import com.ccf.business.etl.service.*;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.component.api.model.flow.FlowInfo;
import com.component.api.model.flow.NodeInfo;
import com.component.api.model.flow.RelationInfo;
import com.ccf.business.etl.flow.ParamManager;
import com.ccf.business.etl.flow.mq.FlowRunningProducer;
import com.ccf.business.etl.mapper.TaskMapper;
import com.ccf.business.etl.model.dto.FlowInfoDTO;
import com.ccf.business.etl.model.dto.NodeDTO;
import com.ccf.business.etl.xxljob.admin.XxlJobAdminApi;
import com.ccf.business.etl.xxljob.model.XxlJobInfo;
import com.common.model.exception.FlowAnalysisException;
import com.common.model.exception.ServiceException;
import com.common.model.pojo.Query;
import com.common.model.result.CodeEnum;
import com.common.model.result.PageResult;
import com.service.mybatis.pojo.SuperEntity;
import com.service.mybatis.service.impl.SuperServiceImpl;
import com.service.mybatis.util.Condition;
import lombok.extern.slf4j.Slf4j;
import org.apache.logging.log4j.core.util.CronExpression;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.time.ZoneId;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.stream.Collectors;

/**
 * 任务调度表
 *
 * @author zhaoj
 * @date 2021-12-01 15:15:53
 */
@Slf4j
@Service
public class TaskServiceImpl extends SuperServiceImpl<TaskMapper, Task> implements ITaskService {

    @Autowired
    private XxlJobAdminApi xxlJobAdminApi;

    @Autowired
    private IFlowService flowService;

    @Autowired
    private IJobService jobService;

    @Autowired
    private INodeService nodeService;

    @Autowired
    private FlowRunningProducer flowRunningProducer;

    @Autowired
    private ParamManager paramManager;

    @Autowired
    private IFlowLogService flowLogService;
    /**
    * 列表
    * @param query
    * @return
    */
    @Override
    public PageResult<Task> findList(Query query){
        Page<Task> page = page(Condition.getPage(query), Condition.getQueryWrapper(query));
        return PageResult.<Task>builder().datas(page.getRecords()).code(CodeEnum.SUCCESS.getCode()).count(page.getTotal()).build();
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public boolean submit(Task task)  {
        String taskId = IdUtil.fastSimpleUUID();
        Flow flow = flowService.getById(task.getFlowId());
        if (ObjectUtil.isNull(flow)) {
            throw new ServiceException("数据未发现");
        }
        //拷贝节点数据
        try {
            flowService.copy(task.getFlowId(), taskId);
        } catch (Exception e) {
            e.printStackTrace();
            throw new RuntimeException("流程缺少必要参数，不能调度");
        }

        //获取下一次调度时间
        DateTime date = null;
        if(Task.SCHEDULEMODE_CYCLE.equals(task.getScheduleMode())) {
            date = DateUtil.date(task.getStartDate());
            date.setField(DateField.HOUR_OF_DAY, task.getStartHour());
            date.setField(DateField.MINUTE, task.getStartMin());
            if(System.currentTimeMillis() > date.getTime()) {
                //延后10S
                date = DateUtil.offsetSecond(new Date(), 10);
            }

            //周期调度设置下次运行时间设置下次运行时间
            try {
                CronExpression cronExpression = new CronExpression(task.getCorn());
                task.setNextPlanTime(cronExpression.getNextValidTimeAfter(new Date(date.getTime())).toInstant().atZone(ZoneId.systemDefault()).toLocalDateTime());
            } catch (Exception e) {
                e.printStackTrace();
            }
        } else {
            date = new DateTime();
            task.setNextPlanTime(date.toJdkDate().toInstant().atZone(ZoneId.systemDefault()).toLocalDateTime());
        }

        //保存任务到xxl-job
        XxlJobInfo jobInfo = new XxlJobInfo();
        jobInfo.setScheduleConf(task.getCorn());
        jobInfo.setExecutorHandler("flowTaskJob");
        jobInfo.setJobDesc(task.getName());
        jobInfo.setTriggerNextTime(date.getTime());
        jobInfo.setExecutorParam(taskId);

        String xxlJobId = xxlJobAdminApi.addJobInfo(jobInfo);
        if(StrUtil.isNotEmpty(xxlJobId)) {
            //拷贝任务
            task.setId(taskId);
            task.setData(flow.getData());
            task.setStatus(Task.STATUS_WAITTING);
            task.setXxlJobId(xxlJobId);
            task.setCreator("1");

            //保存数据
            boolean result =  this.save(task);
            //如果是单次调度则立即执行
            if(Task.SCHEDULEMODE_ONCE.equals(task.getScheduleMode())) {
                this.running(task);
            //启动调度任务
            } else {
                xxlJobAdminApi.startJobInfo(xxlJobId);
            }
            return result;
        }

        return false;
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public boolean deleteBatch(List<String> ids) {
        //获取所有数据
        List<Task> tasks = this.list(Wrappers.<Task>lambdaQuery().in(SuperEntity::getId, ids));
        //删除调度表
        for (Task t : tasks) {
            xxlJobAdminApi.removeJobInfo(t.getXxlJobId());
        }
        //删除Job表
        jobService.remove(Wrappers.<Job>lambdaQuery().in(Job::getTaskId, ids));
        return this.removeByIds(ids);
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public boolean start(List<String> ids) {
        List<Task> tasks = this.list(Wrappers.<Task>lambdaQuery().in(SuperEntity::getId, ids));
        for (Task t : tasks) {
            if(Task.SCHEDULEMODE_CYCLE.equals(t.getScheduleMode())) {
                xxlJobAdminApi.startJobInfo(t.getXxlJobId());
                t.setStatus(Task.STATUS_WAITTING);
                this.updateById(t);
            } else {
                this.running(t);
            }
        }
        return true;
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public boolean stop(List<String> ids) {
        List<Task> tasks = this.list(Wrappers.<Task>lambdaQuery().in(SuperEntity::getId, ids));
        for (Task t : tasks) {
            if(Task.SCHEDULEMODE_CYCLE.equals(t.getScheduleMode())) {
                xxlJobAdminApi.stopJobInfo(t.getXxlJobId());
                t.setStatus(Task.STATUS_STOPPING);
                this.updateById(t);
            } else {
                if(t.getStatus().equals(Task.STATUS_RUNNING)) {
                    //修改任务为以完成
                    t.setStatus(Task.STATUS_STOPPING);
                    //修改JOB为已完成
                    Job job = jobService.list(Wrappers.<Job>lambdaQuery().eq(Job::getTaskId, t.getId()).orderByDesc(SuperEntity::getCreateTime)).stream().findFirst().orElse(null);
                    if(job.getStatus().equals(Task.STATUS_RUNNING)) {
                        job.setStatus(Task.STATUS_STOPPING);
                        jobService.updateById(job);
                    }
                    this.updateById(t);
                }
            }
        }
        return true;
    }

    @Override
    public boolean trigger(Task task) {
        task = this.getById(task.getId());
        if(Task.SCHEDULEMODE_CYCLE.equals(task.getScheduleMode())) {
            boolean b = xxlJobAdminApi.triggerJobInfo(task.getXxlJobId(), task.getId());
            if(b) {
                task.setStatus(Task.STATUS_RUNNING);
                boolean result = this.updateById(task);
                return  result;
            }
        } else {
            this.running(task);
            return true;
        }
        return false;
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public void running(Task task) {
        //保存任务实例
        Job job = new Job();
        job.setFlowId(task.getFlowId());
        job.setStartTime(new Date());
        job.setTaskId(task.getId());
        job.setStatus(Task.STATUS_RUNNING);
        jobService.save(job);

        //更新Task为运行中
        task.setStatus(Task.STATUS_RUNNING);
        this.updateById(task);

        try {
            //这里是获取任务数据
            String data = task.getData();
            FlowInfoDTO flowInfoDTO = JSON.parseObject(data, FlowInfoDTO.class);
            FlowInfo flowInfo = new FlowInfo();
            //这里改成任务id
            flowInfo.setFlowId(job.getId());
            //这里改成任务名称
            flowInfo.setFlowName(task.getName());
            flowInfo.setEdges(flowInfoDTO.getEdges().stream()
                    .map(e->
                            RelationInfo.builder().id(e.getId()).source(e.getSource()).target(e.getTarget()).build()
                    ).collect(Collectors.toList()));
            List<NodeInfo> nodeInfos = Lists.newArrayList();
            Map<String, Object> paramMap = Maps.newHashMap();
            paramMap.put("flow_id", flowInfoDTO.getId());
            List<Node> nodes = nodeService.listByMap(paramMap);
            for(Node node : nodes){
                try {
                    NodeInfo nodeInfo = paramManager.getHandler(new NodeDTO(node)).get();
                    nodeInfos.add(nodeInfo);
                } catch (Exception e) {
                    e.printStackTrace();
                    throw new FlowAnalysisException(node.getName() + "参数错误->"+e.getMessage());
                }
            }
            flowInfo.setNodes(nodeInfos);
            //提交运行
            flowRunningProducer.running(flowInfo);
        } catch (Exception e) {
            FlowLog flowLog = new FlowLog();
            flowLog.setFlowId(job.getId());
            flowLog.setFlowName(task.getName());
            flowLog.setLogLevel("error");
            flowLog.setLogTime(new Date());
            flowLog.setMsg(e.getMessage());
            flowLogService.save(flowLog);
            //手动完成
            this.finish(job.getId(), Task.STATUS_FAIL);
        }

    }

    @Override
    public void finish(String jobId, String status) {
        Job job = jobService.getById(jobId);
        if (job == null || job.getStatus().equals(Task.STATUS_STOPPING)){
            return;
        }
        job.setStatus(status);
        job.setEndTime(new Date());
        //计算用时
        long time = job.getEndTime().getTime() - job.getStartTime().getTime();
        job.setElapsedTime(time);
        jobService.updateById(job);

        //获取任务状态并更新
        Task task = this.getById(job.getTaskId());
        if (Task.SCHEDULEMODE_ONCE.equals(task.getScheduleMode())) {
            //处理单次调度任务
            if(!status.equals(Task.STATUS_FAIL)) {
                task.setStatus(Task.STATUS_FINISH);
            } else {
                task.setStatus(Task.STATUS_FAIL);
            }
        } else {
            if(status.equals(Task.STATUS_FAIL)) {
                task.setStatus(Task.STATUS_FAIL);
                //运行失败停止调度
                xxlJobAdminApi.removeJobInfo(task.getXxlJobId());
            } else if(status.equals(Task.STATUS_FINISH)) {
                task.setStatus(Task.STATUS_WAITTING);
            }

            //周期调度设置下次运行时间设置下次运行时间
            try {
                CronExpression cronExpression = new CronExpression(task.getCorn());
                task.setNextPlanTime(cronExpression.getNextValidTimeAfter(new Date()).toInstant().atZone(ZoneId.systemDefault()).toLocalDateTime());

                //处理周期任务
                if(ObjectUtil.isNotNull(task.getEndDate()) && Date.from(task.getNextPlanTime().atZone(ZoneId.systemDefault()).toInstant()).getTime() > DateUtil.endOfDay(Date.from(task.getEndDate().atZone(ZoneId.systemDefault()).toInstant())).getTime()) {
                    if(!task.getStatus().equals(Task.STATUS_FAIL)) {
                        task.setStatus(Task.STATUS_FINISH);
                    }
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
        this.updateById(task);
    }

    @Override
    public List<Task> getTaskByFlowIds(Set<String> ids) {
        LambdaQueryWrapper<Task> eq = Wrappers.<Task>lambdaQuery()
                .in(CollUtil.isNotEmpty(ids), Task::getFlowId, ids);
        List<Task> list = list(eq);
        return list;
    }

}
